package cn.csc.module.pay.job.mq;

import cn.csc.framework.common.pojo.PageResult;
import cn.csc.framework.common.util.json.JsonUtils;
import cn.csc.framework.tenant.core.context.TenantContextHolder;
import cn.csc.framework.tenant.core.job.TenantJob;
import cn.csc.module.pay.controller.app.mqrecord.vo.AppPayMqRecordPageReqVO;
import cn.csc.module.pay.dal.dataobject.jzborder.JzbOrderDO;
import cn.csc.module.pay.dal.dataobject.mqrecord.PayMqRecordDO;
import cn.csc.module.pay.dal.mysql.jzborder.JzbOrderMapper;
import cn.csc.module.pay.dal.mysql.mqrecord.PayMqRecordMapper;
import cn.csc.module.pay.message.JzbOrderMessage;
import cn.csc.module.pay.mq.producer.JzbOrderProducer;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

/**
 * mq消息补偿生产
 * @author liuchuang
 * @version 1.0
 * @date 2023/11/27 16:45
 * @modify
 */
@Component
@TenantJob // 多租户
@Slf4j
public class MqProduceJob {

    @Resource
    private JzbOrderProducer jzbOrderProducer;

    @Resource
    private PayMqRecordMapper payMqRecordMapper;

    @Resource
    private JzbOrderMapper jzbOrderMapper;

    /**
     * 冻结充值消息补偿
     * job参数格式：orderId,tenantId
     */
    @XxlJob("sendJzbOrderMsgJob")
    public ReturnT<String> sendJzbOrderMsgJob() {
        String params = XxlJobHelper.getJobParam();
        log.info("sendJzbOrderMsgJob 输入的任务参数: {}", params);
        if (!StringUtils.hasLength(params)){
            XxlJobHelper.handleFail("输入的任务参数为空！");
            return new ReturnT<>(500, "输入的任务参数为空");
        }
        String[] split = params.split(",");
        TenantContextHolder.setTenantId(Long.valueOf(split[1]));
        PayMqRecordDO paramDo = PayMqRecordDO.builder().orderId(new BigInteger(split[0]))
                .busiType("1")           // 冻结充值
                .deliverStatus("0")      //未投递
                .consumeStatus("0")      // 未消费
                .build();
        PayMqRecordDO recordDO = payMqRecordMapper.selectOne(paramDo);
        log.info("sendJzbOrderMsgJob recordDO:{}", recordDO);

        JzbOrderMessage jzbOrderMessage = JSONObject.parseObject(recordDO.getMessage(), JzbOrderMessage.class);

        // 返回true为发送成功
        boolean sendMsg = jzbOrderProducer.sendJzbOrderMessage(jzbOrderMessage);
        log.info("sendMsg: {}", sendMsg);
        if (sendMsg){
            recordDO.setDeliverStatus("1");
            payMqRecordMapper.updateById(recordDO);
            XxlJobHelper.handleSuccess("消息发送成功");
            return new ReturnT<>(String.format("消息发送成功: %s ", recordDO));
        }
        XxlJobHelper.handleFail(String.format("消息发送失败，消息内容： %s ", recordDO));
        return new ReturnT<>(500, String.format("消息发送失败，消息内容： %s ", recordDO));

    }

    /**
     * 补偿见证宝订单流水
     * job参数格式：jzbOrderId,tenantId
     */
    @XxlJob("compensateJzbOrderMsgJob")
    public ReturnT<String> compensateJzbOrderMsgJob(){
        String params = XxlJobHelper.getJobParam();
        log.info("compensateJzbOrderMsgJob 输入的任务参数param: {}", params);
        if (!StringUtils.hasLength(params)){
            XxlJobHelper.handleFail("输入的任务参数为空！");
            return new ReturnT<>(500, "输入的任务参数为空");
        }
        String[] split = params.split(",");
        TenantContextHolder.setTenantId(Long.valueOf(split[1]));
        JzbOrderDO jzbOrderDO = jzbOrderMapper.selectById(new BigInteger(split[0]));
        log.info("compensateJzbOrderMsgJob jzbOrderDO: {}", jzbOrderDO);
        Assert.notNull(jzbOrderDO , "见证宝订单不能为空");

        // 日志流水表
        PayMqRecordDO recordDO = new PayMqRecordDO();
        recordDO.setBusiType("1");  // 冻结充值
        recordDO.setOrderId(jzbOrderDO.getPayOrderId());
        recordDO.setJzbOrderId(jzbOrderDO.getId());
        recordDO.setChannel("jzbOrder-out-0");
        JzbOrderMessage jzbOrderMessage = new JzbOrderMessage(jzbOrderDO.getId(),jzbOrderDO.getFreezeRechargeNo(),"1");
        recordDO.setMessage(JsonUtils.toJsonString(jzbOrderMessage));
        recordDO.setDeliverStatus("0");
        // 业务流水表，保障消息投递，防止重复消费
        payMqRecordMapper.insert(recordDO);

        log.info("compensateJzbOrderMsgJob recordDO: {}", recordDO);
        return new ReturnT<>(String.format("补偿见证宝订单流水: %s ", recordDO));
    }

    /**
     * 见证宝订单消息消费失败，补偿发送
     */
    @XxlJob("compensateConsumeFailJob")
    public ReturnT<String> compensateConsumeFailJob(){
        AppPayMqRecordPageReqVO reqVO = new AppPayMqRecordPageReqVO();
        reqVO.setPageNo(1);
        reqVO.setPageSize(30);
        reqVO.setDeliverStatus("1");
        reqVO.setConsumeStatus("0");

        PageResult<PayMqRecordDO> pageResult = payMqRecordMapper.selectPage(reqVO);
        Long total = pageResult.getTotal();
        if (total < 1){
            log.info("[compensateConsumeFailJob]无消费失败的见证宝订单消息");
            XxlJobHelper.handleSuccess(String.format("补偿发送消费失败的见证宝订单消息 %s 条", 0));
            return new ReturnT<>("无消费失败的见证宝订单消息");
        }

        // 计算总页数
        int pages = 1;
        // 总行数 大于 每页条数
        if (total.intValue() > reqVO.getPageSize()) {
            if ((total.intValue() % reqVO.getPageSize()) == 0) {
                pages = total.intValue() / reqVO.getPageSize();
            } else {
                pages = (total.intValue() / reqVO.getPageSize()) + 1;
            }
        }

        AtomicInteger count = new AtomicInteger(0);
        // 第一页
        pageResult.getList().forEach(mq -> {
            JzbOrderMessage jzbOrderMessage = JSONObject.parseObject(mq.getMessage(), JzbOrderMessage.class);
            // 返回true为发送成功
            boolean sendMsg = jzbOrderProducer.sendJzbOrderMessage(jzbOrderMessage);;
            if (sendMsg){
                count.incrementAndGet();
                mq.setDeliverNum(mq.getDeliverNum()+1);
                payMqRecordMapper.updateById(mq);
            }
        });

        // 第二页 至 最后一页, 前后都包含
        IntStream.rangeClosed(2, pages).forEach(p -> {
            // 设置页码
            reqVO.setPageNo(p);
            // 分页查询
            PageResult<PayMqRecordDO> result = payMqRecordMapper.selectPage(reqVO);
            result.getList().forEach(mq -> {
                JzbOrderMessage jzbOrderMessage = JSONObject.parseObject(mq.getMessage(), JzbOrderMessage.class);
                // 返回true为发送成功
                boolean sendMsg = jzbOrderProducer.sendJzbOrderMessage(jzbOrderMessage);;
                if (sendMsg){
                    count.incrementAndGet();
                    mq.setDeliverNum(mq.getDeliverNum()+1);
                    payMqRecordMapper.updateById(mq);
                }
            });
        });
        log.info("[compensateConsumeFailJob] {}", String.format("补偿发送消费失败的见证宝订单消息 %s 条", count.get()));
        XxlJobHelper.handleSuccess(String.format("补偿发送消费失败的见证宝订单消息 %s 条", count.get()));
        return new ReturnT<>(String.format("补偿发送消费失败的见证宝订单消息 %s 条", count.get()));
    }



    @XxlJob("testSelectJob")
    public ReturnT<String> testSelectJob() {
        String orderId = XxlJobHelper.getJobParam();
        log.info("testSelectJob 输入的任务参数: {}", orderId);
        if (!StringUtils.hasLength(orderId)){
            return new ReturnT<>(500, "输入的任务参数为空");
        }
        PayMqRecordDO paramDo = PayMqRecordDO.builder()
                .orderId(new BigInteger(orderId))
                .busiType("1")
                .build();
        PayMqRecordDO recordDO = payMqRecordMapper.selectOne(paramDo);
        log.info("testSelectJob recordDO:{}", recordDO);
        XxlJobHelper.handleSuccess("执行成功！");
        return new ReturnT<>(200, "执行成功！");
    }


}
